package com.atguigu.day08

import java.util.Properties

import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object $04_KafkaSource {
  /**
    * sparkstreaming消费kafka数据的时候,rdd的分区数 = topic的分区数,而且是动态感知的,也就是说当topic的分区数改变的时候rdd的分区数也会改变
    * @param args
    */
  def main(args: Array[String]): Unit = {
    //1、创建StreamingContext对象
    val conf = new SparkConf().setMaster("local[4]").setAppName("test")
    // 500ms
    val ssc = new StreamingContext( conf, Seconds(5) )
    ssc.sparkContext.setLogLevel("error")

    //2、从数据源读取数据
    val topics = Array("wordcount")

    //设置kafka消费者相关参数
    val props = Map[String,Object](
      //指定key的反序列化器
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      //指定value的反序列化器
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      //指定集群地址
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      //指定消费组id
      "group.id" -> "g11",
      //指定消费组第一次消费的时候从哪个位置开始消费
      "auto.offset.reset" -> "earliest"
    )
    val ds = KafkaUtils.createDirectStream[String,String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String,String](topics,props))

    //3、数据处理
    //取出消息
    ds.map(record => record.value())
      .foreachRDD(rdd => {

        val rdd2 = rdd.flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_)
        println(rdd2.partitions.length)
        rdd2.collect().foreach(println(_))
      })

    //4、结果展示

    //5、启动
    ssc.start()
    //6、阻塞
    ssc.awaitTermination()
  }
}
